package com.feidee.fd.sml.algorithm.forecast

import java.util.ArrayList
import java.{lang, util}

import com.feidee.fd.sml.algorithm.util.{Log, ToolClass}
import org.apache.spark.sql.{DataFrame, SparkSession}

import scala.collection.JavaConversions._
import scala.collection.JavaConverters._
import scala.collection.mutable.ArrayBuffer
import scala.collection.{JavaConversions, mutable}

/**
  * @Author: xiongjun, caokaizhi, songhc
  * @Date: 2019/5/6 14:09
  */
trait Forecast[A] extends Log {
  val tool = new ToolClass
  val spark: SparkSession
  val paths: Seq[String]
  val fieldInfo: Map[String, Array[String]]


  protected implicit def toMap(map: mutable.Map[String, Any]): Map[String, Any] = map.toMap

  protected implicit def toMaps(maps: Seq[mutable.Map[String, Any]]): Seq[Map[String, Any]] = maps.map(_.toMap)

  def load(): Array[A]

  def predict(maps: Seq[Map[String, Any]]): Array[Map[String, Any]]

  def predictJavaList(javaList: util.List[util.Map[String, Any]]): util.List[util.Map[String, Any]] = {
    val scalaRes = predict(JavaConversions.asScalaBuffer(javaList.map(JavaConversions.mapAsScalaMap)).toSeq)
    scalaRes.map(JavaConversions.mapAsJavaMap(_)).toList
  }

  def predictJavaMap(javaMap: util.Map[String, util.Map[String, Any]]): util.Map[String, util.Map[String, Any]] = {
    val keys = new ArrayBuffer[String](javaMap.size())
    val maps = new ArrayBuffer[util.Map[String, Any]](javaMap.size())

    javaMap.entrySet().foreach(e => {
      keys += e.getKey
      maps += JavaConversions.mapAsScalaMap(e.getValue)
    })

    val resList = predictJavaList(maps)
    var resMap = Map[String, util.Map[String, Any]]()

    for (i <- resList.indices) {
      resMap += (keys(i) -> resList(i))
    }

    JavaConversions.mapAsJavaMap(resMap)
  }

  /**
    * 返回预测标签 topn 项和对应概率值
    *
    * @param prediction 已经预测好的原始 DataFrame
    * @param topn       as it indicates
    * @return
    */
  def computeProbability(prediction: DataFrame, topn: Int): DataFrame

  /**
    * 用加载好的模型预测并返回预测概率值
    *
    * @param maps
    * @param topn
    * @return
    */
  def predictProbabilityMaps(maps: Seq[Map[String, Any]], topn: Int, preModels: Array[Forecast[Any]]): Array[Map[String, Double]]

  /**
    * 用加载好的回归模型预测并返回实数值
    *
    * @param maps
    * @return
    */
  def getPredictionMaps(maps: Seq[Map[String, Any]]): Array[Double]

  /**
    * 提供给sml-model调用的方法，用加载好的分类模型进行不含多分类特征的预测
    *
    * @param javaMaps
    * @return
    */
  def predictProbabilityJavaList(javaMaps: util.List[util.Map[String, Any]], topn: Int, preModels: Array[Forecast[Any]]): util.List[util.Map[lang.String, lang.Double]] = {
    val scalaRes = predictProbabilityMaps(
      JavaConversions.asScalaBuffer(
        javaMaps.map(
          JavaConversions.mapAsScalaMap)).toSeq, topn, preModels)

    JavaConversions.seqAsJavaList(
      scalaRes.map(r => {
        JavaConversions.mapAsJavaMap(r.map(m => {
          (m._1, lang.Double.valueOf(m._2))
        }))
      }))
  }

  /**
    * 提供给sml-model调用的方法，用加载好的分类模型进行多分类特征的预测
    */
  def predictProbabilityJavaMap(javaMap: util.Map[String, util.Map[String, Any]], topn: Int = 0, preModels: Array[Forecast[Any]]):
  util.Map[String, util.Map[lang.String, lang.Double]] = {
    val keys = new ArrayBuffer[String](javaMap.size())
    val maps = new ArrayBuffer[Map[String, Any]](javaMap.size())

    javaMap.entrySet().foreach(e => {
      keys += e.getKey
      maps += JavaConversions.mapAsScalaMap(e.getValue)
    })

    val scalaResList = predictProbabilityMaps(maps, topn, preModels)
    var scalaResMap = Map[String, Map[String, Double]]()

    for (i <- scalaResList.indices) {
      scalaResMap += (keys(i) -> scalaResList(i))
    }

    JavaConversions.mapAsJavaMap(
      scalaResMap.map(r => {
        (r._1, JavaConversions.mapAsJavaMap(r._2.map(m => (m._1, lang.Double.valueOf(m._2)))))
      }))
  }

  /**
    * 提供给sml-model调用的方法，用加载好的回归模型进行不含多分类特征的预测
    */
  def predictionRegJavaList(javaMaps: util.List[util.Map[String, Any]]): util.List[lang.Double] = {
    val scalaRes = getPredictionMaps(
      JavaConversions.asScalaBuffer(
        javaMaps.map(
          JavaConversions.mapAsScalaMap)).toSeq)

    scalaRes.toList.map(r => lang.Double.valueOf(r))
  }

  /**
    * 提供给sml-model调用的方法，用加载好的回归模型进行多分类特征的预测
    */
  def predictionRegJavaMap(javaMap: util.Map[String, util.Map[String, Any]]):util.Map[String, lang.Double] = {
    val keys = new ArrayBuffer[String](javaMap.size()) //TODO 所有推广计划id集合
    val maps = new ArrayBuffer[Map[String, Any]](javaMap.size()) //TODO 所有推广计划id的特征集合

    javaMap.entrySet().foreach(e => {
      keys += e.getKey
      maps += JavaConversions.mapAsScalaMap(e.getValue)
    })

    val scalaResList = getPredictionMaps(maps) //TODO 拿到所有推广计划的一个单一值集合,结果例:List(38.3315, 34.5845, 33.6451),注意此处的list为scala集合
    val list = scalaResList.toList.map(r => lang.Double.valueOf(r)) //TODO 将结果中的单值转为java中的Double类型

    var scalaResMap = Map[String, lang.Double]()

    for (i <- scalaResList.indices) {
      scalaResMap += (keys(i) -> list(i)) //TODO 存放所有的推广计划id的预测值，结果例:(19457->38.3315,18848->34.5845,19444->33.6451)
    }

   scalaResMap.asJava //TODO 返回java中的Map集合

  }

  def warmup(javaMaps: util.List[util.Map[String, Any]]): util.List[util.Map[lang.String, lang.String]] = {
    val scalaRes = predict(
      JavaConversions.asScalaBuffer(
        javaMaps.map(
          JavaConversions.mapAsScalaMap)).toSeq)

    JavaConversions.seqAsJavaList(
      scalaRes.map(r => {
        JavaConversions.mapAsJavaMap(r.map(m => {
          (m._1, m._2.toString)
        }))
      }))
  }
  def intermediateResult(javaMaps: util.List[util.Map[String, Any]]): util.List[util.Map[lang.String, lang.Object]]={
    val scalaRes = predict(
      JavaConversions.asScalaBuffer(
        javaMaps.map(
          JavaConversions.mapAsScalaMap)).toSeq)
    val result = JavaConversions.seqAsJavaList(
      scalaRes.map(r => {
        JavaConversions.mapAsJavaMap(r.map(m => {
          if(m._2.isInstanceOf[Stream[Any]]){
            (m._1,m._2.asInstanceOf[Stream[Any]].toArray)
          }else{
            (m._1,m._2.asInstanceOf[Object])
          }
        }))
      }))
    result
  }

}

object Forecast {
  final val LABEL2PROBABILITY_COL = "label2Probability"
  final val PROBABILITY_REG = "^probability\\((.*)\\)$".r
  final val MODEL_TYPE_SPARK = "spark"
  final val MODEL_TYPE_PMML = "pmml"
  final val MODEL_TYPE_MLEAP = "mleap"
  final val MODEL_TYPE_TF = "tensorflow"
  final val SUPPORTED_MODEL_TYPES = Array(MODEL_TYPE_SPARK, MODEL_TYPE_PMML, MODEL_TYPE_MLEAP, MODEL_TYPE_TF)
}
